package com.jml;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.jml.utils.RedisUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;


@RestController
@Slf4j
public class KafkaController {

    /**
     * 注入kafkaTemplate
     */
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Autowired
    private RedisUtils redisUtils;

    /**
     * 发送消息的方法
     *
     * @param key  推送数据的key
     * @param data 推送数据的data
     */
    private void send(String key, String data) {
        // topic 名称 key data 消息数据
        kafkaTemplate.send("jml", key, data);

    }
    // test 主题 1 my_test 3

    @RequestMapping("/kafka")
    public String testKafka() {
        int iMax = 6;
        for (int i = 1; i < iMax; i++) {
            send("key" + i, "data" + i);
        }
        return "success";
    }

    public static void main(String[] args) {
        SpringApplication.run(KafkaController.class, args);
    }

    /**
     * 消费者使用日志打印消息
     */

    @KafkaListener(topics = "jml-topic")
    public void receive(ConsumerRecord<?, ?> consumer) {
        String index = consumer.offset() + "" + consumer.value();
        log.info(">topic名称:{},,key:{},分区位置:{},, 下标{}<", consumer.topic(), consumer.key(), consumer.partition(), index);
        String json = (String) consumer.value();
        JSONObject jsonObject = JSONObject.parseObject(json);
        String sqlType = jsonObject.getString("type");
        JSONArray data = jsonObject.getJSONArray("data");
        JSONObject userObject = data.getJSONObject(0);
        String id = userObject.getString("id");
        String database = jsonObject.getString("database");
        String table = jsonObject.getString("table");
        String key = database + "_" + table + "_" + id;
        if ("UPDATE".equals(sqlType) || "INSERT".equals(sqlType)) {
            redisUtils.setString(key, userObject.toJSONString());
            return;
        }
        if ("DELETE".equals(sqlType)) {
            redisUtils.deleteKey(key);
        }
    }

}
